package monitor

import (
	"context"
	"cloud_worm/worm/client/monitor/kafka"
	"sync"
	"github.com/golang/glog"
	"fmt"
	"cloud_worm/util/proto/ujson"
)

const (
	CmdTopic = "afocus_log_collect"
	LogTopic = "afocus_claw_log_collect"

	kafkaBroker = "blackstonekf1.service.mogujie.org:9092,blackstonekf2.service.mogujie.org:9092,blackstonekf3.service.mogujie.org:9092"
)

var producerPool = &ProducerPool{}

type ProducerClient struct {
	kafka.KafProducer
	count 			int
}
type ProducerPool struct {
	producers 	map[string]*ProducerClient
	mux 		sync.Mutex
}

func init()  {
	producerPool.producers = make(map[string]*ProducerClient)
}

func (p *ProducerPool) Get(topic string) (kafka.KafProducer, error) {
	p.mux.Lock()
	defer p.mux.Unlock()

	if _, exist := p.producers[topic]; exist == false {
		// 没有该kafka的producer，则创建一个
		ctx, cancel := context.WithCancel(context.Background())
		producer := kafka.NewAsyncProducer(topic, kafkaBroker, cancel)
		if err := producer.DialKafka(ctx); err != nil {
			return nil, err
		}
		producer.ErrLog(ctx)
		glog.Info(fmt.Sprintf("create kafka producer for topic: %s", topic))

		p.producers[topic] = &ProducerClient{
			KafProducer: producer,
			count: 0,
		}
	}

	p.producers[topic].count++
	return p.producers[topic].KafProducer, nil
}

func (p *ProducerPool) Put(topic string) {
	p.mux.Lock()
	defer p.mux.Unlock()

	if _, exist := p.producers[topic]; exist {
		p.producers[topic].count--
		if p.producers[topic].count == 0 {
			p.producers[topic].KafProducer.Close()
			delete(p.producers, topic)
			glog.Info(fmt.Sprintf("drop kafka producer for topic: %s", topic))
		}
	}
	return
}

// 销毁结束时，调用该接口执行资源释放
// 暂时不考虑该接口
//func (p *ProducerPool) Clear()  {
//	p.mux.Lock()
//	defer p.mux.Unlock()
//
//	for _, v := range p.producers {
//		v.Close()
//		v.count = 0
//	}
//
//	p.producers = make(map[string]*ProducerClient)
//	return
//}

var PushMsgC chan []byte

func PushMonitorErrInfo(err error)  {
	if PushMsgC != nil {
		errInfo := ujson.ErrMsg{
			Err: err.Error(),
		}
		if p, err := errInfo.MarshalJSON(); err == nil {
			pushMsg := ujson.PushMsg{
				Type: ujson.PushErrMsg,
				Payload: p,
			}
			if pp, err := pushMsg.MarshalJSON(); err == nil {
				msg := ujson.ProtoMsg{
					Type: ujson.ProtoMsgPush,
					Content: pp,
				}
				if buf, err := msg.MarshalJSON(); err == nil {
					PushMsgC <- buf
				}
			}
		}
	}

	return
}
